package shujia;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.*;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.util.Bytes;
import org.junit.After;
import org.junit.Before;
import org.junit.Test;

import java.io.BufferedReader;
import java.io.FileReader;
import java.io.IOException;
import java.util.ArrayList;
import java.util.List;

public class Demo2API {

    Connection conn;
    Admin admin;

    @Before
    public void createConn() throws IOException {
        // 1、创建一个配置文件
        Configuration conf = HBaseConfiguration.create();
        // 配置ZK的地址，通过ZK可以找到HBase
        conf.set("hbase.zookeeper.quorum", "master:2181,node1:2181,node2:2181");

        // 2、创建连接
        conn = ConnectionFactory.createConnection(conf);
        // 3、创建Admin对象
        admin = conn.getAdmin();
    }

    @Test
    /**
     * create table
     *
     * student info
     */
    public void createTable() throws IOException {

        HTableDescriptor student = new HTableDescriptor(TableName.valueOf("student"));
        HColumnDescriptor info = new HColumnDescriptor("info");
//        info.setTimeToLive(100);// 对列簇进行配置
        student.addFamily(info);
        admin.createTable(student);
    }

    @Test
    /**
     * drop table
     */
    public void dropTable() throws IOException {
        TableName test2 = TableName.valueOf("student");
        // 判断表是否存在
        if (admin.tableExists(test2)) {
            admin.disableTable(test2);
            admin.deleteTable(test2);
        }

    }

    @Test
    /**
     * 修改表结构
     * 针对test表 将其info列簇的ttl设置为10000，并增加一个新的列簇cf1
     */
    public void modifyTable() throws IOException {
        TableName test = TableName.valueOf("test");
        // 获取表原有的结构
        HTableDescriptor tableDescriptor = admin.getTableDescriptor(test);
        // 在表原有的结构中 修改列簇的属性
        HColumnDescriptor[] columnFamilies = tableDescriptor.getColumnFamilies();
        // 遍历表中原有的列簇
        for (HColumnDescriptor cf : columnFamilies) {
            // 对原有的info列簇进行修改
            if ("info".equals(cf.getNameAsString())) {
                cf.setTimeToLive(10000);
            }
        }
        // 新增一个列簇
        HColumnDescriptor cf1 = new HColumnDescriptor("cf1");
        tableDescriptor.addFamily(cf1);

        admin.modifyTable(test, tableDescriptor);
    }

    @Test
    /**
     * list 列出所有的表
     */
    public void list() throws IOException {
        TableName[] tableNames = admin.listTableNames();
        for (TableName name : tableNames) {
            System.out.println(name.getNameAsString());
        }
    }

    @Test
    /**
     * put 插入一条数据
     */
    public void put() throws IOException {
        Table testAPI = conn.getTable(TableName.valueOf("testAPI"));

        Put put = new Put("0002".getBytes());
        // 相当于插入一列（一个cell）数据
        put.addColumn("cf1".getBytes(), "name".getBytes(), "李四".getBytes());
        put.addColumn("cf1".getBytes(), "age".getBytes(), "23".getBytes());
        put.addColumn("cf1".getBytes(), "phone".getBytes(), "18888887".getBytes());

        testAPI.put(put);
    }

    @Test
    /**
     * get 根据rowkey获取一条数据
     */
    public void get() throws IOException {
        Table testAPI = conn.getTable(TableName.valueOf("testAPI"));

        Get get = new Get("0002".getBytes());

        Result rs = testAPI.get(get);
        // 获取rk
        byte[] rk = rs.getRow();
        System.out.println(rk);
        System.out.println(Bytes.toString(rk));
        // 获取cell
        byte[] name = rs.getValue("cf1".getBytes(), "name".getBytes());
        System.out.println(name);
        System.out.println(Bytes.toString(name));

    }

    @Test
    /**
     * delete 删除数据
     */
    public void deleteLine() throws IOException {
        Delete delete = new Delete("1500100001".getBytes());
        Table student = conn.getTable(TableName.valueOf("student"));
        student.delete(delete);
    }

    @Test
    /**
     * putAll 读取学生信息数据并写入HBase的student表
     */
    public void putAll() throws IOException {
        /**
         * 读取学生信息数据
         */

        // Junit 和 main方法运行时的工作路径不一样
        // 这里传入的相对路径要动态调整
        BufferedReader br = new BufferedReader(new FileReader("data/students.txt"));

        // 与HBase中的student表建立连接
        Table student = conn.getTable(TableName.valueOf("student"));

        String line = null;

        // 创建Put的集合
        ArrayList<Put> puts = new ArrayList<>();
        int batchSize = 11;
        while ((line = br.readLine()) != null) {
            // 写入HBase
            String[] splits = line.split(",");
            String id = splits[0];
            String name = splits[1];
            String age = splits[2];
            String gender = splits[3];
            String clazz = splits[4];

            Put put = new Put(id.getBytes());
            byte[] info = "info".getBytes();
            put.addColumn(info, "name".getBytes(), name.getBytes());
            put.addColumn(info, "age".getBytes(), age.getBytes());
            put.addColumn(info, "gender".getBytes(), gender.getBytes());
            put.addColumn(info, "clazz".getBytes(), clazz.getBytes());


            // 每条数据都会执行一次，效率很慢
//            student.put(put);

            // 将每个Put对象加入puts集合
            puts.add(put);
            // 当puts集合的大小同batchSize大小一致时，则调用HTable的put方法进行批量写入
            if (puts.size() == batchSize) {
                student.put(puts);
                // 清空集合
                puts.clear();
            }

        }
        System.out.println(puts.isEmpty());
        System.out.println(puts.size());
        // 当batchSize的大小同数据的条数不成整比的时候 可能会造成最后几条数据未被写入
        // 手动去判断puts集合是否为空，不为空则将其写入HBase
        if (!puts.isEmpty()) {
            student.put(puts);
        }


        br.close();


    }

    @Test
    /**
     * scan 获取一组数据
     * 读取student表
     */
    public void getScan() throws IOException {
        Table student = conn.getTable(TableName.valueOf("student"));

        // scan可以指定rowkey的范围进行查询，或者是限制返回的条数
        Scan scan = new Scan();
        scan.withStartRow("1500100100".getBytes());
        scan.withStopRow("1500100111".getBytes());
        scan.setLimit(10);

        for (Result rs : student.getScanner(scan)) {
            String id = Bytes.toString(rs.getRow());
            String name = Bytes.toString(rs.getValue("info".getBytes(), "name".getBytes()));
            String age = Bytes.toString(rs.getValue("info".getBytes(), "age".getBytes()));
            String gender = Bytes.toString(rs.getValue("info".getBytes(), "gender".getBytes()));
            String clazz = Bytes.toString(rs.getValue("info".getBytes(), "clazz".getBytes()));

            System.out.println(id + "," + name + "," + age + "," + gender + "," + clazz);
        }

    }

    @Test
    /**
     * CellUtil
     */
    public void scanWithCellUtil() throws IOException {
        Table student = conn.getTable(TableName.valueOf("student"));

        // scan可以指定rowkey的范围进行查询，或者是限制返回的条数
        Scan scan = new Scan();
        scan.withStartRow("1500100990".getBytes());
//        scan.withStopRow("1500100111".getBytes());

        for (Result rs : student.getScanner(scan)) {
            String id = Bytes.toString(rs.getRow());
            System.out.print(id + " ");
            // 将一条数据的所有的cell列举出来
            // 使用CellUtil从每一个cell中取出数据
            // 不需要考虑每条数据的结构
            List<Cell> cells = rs.listCells();
            for (Cell cell : cells) {
                String value = Bytes.toString(CellUtil.cloneValue(cell));
                System.out.print(value + " ");
            }
            System.out.println();

        }
    }


    @After
    public void close() throws IOException {
        admin.close();
        conn.close();
    }

}
